package org.study.module.hadoop;

import org.study.module.common.AbstractConfigBuilder;
import org.study.module.common.ModuleCommon;

import java.util.ArrayList;
import java.util.List;

/**
 * MapReduce配置构建类
 * mapred-site.xml
 */
public class MapReduceConfigBuilder extends AbstractConfigBuilder {

   

    /**
     * 指定 MapReduce 运行框架  yarn集群模式
     */
    public MapReduceConfigBuilder mapreduceFramework() {
        return mapreduceFramework("yarn");
    }

    /**
     * 指定 MapReduce 运行框架（本地模式 vs 集群模式）
     *
     * @param name local：本地模式（单机测试）,yarn：YARN 集群模式（生产默认）
     */
    public MapReduceConfigBuilder mapreduceFramework(String name) {
        addMessage("指定 MapReduce 运行框架");
        addProperty("mapreduce.framework.name", name);
        return this;
    }

    /**
     * YARN HA配置：指定YARN的RM逻辑服务名和地址
     * YARN HA的逻辑服务名（与yarn-site.xml中一致）
     * @param address 逻辑服务名
     */
    public MapReduceConfigBuilder resourceManagerAddress(String address) {
        addMessage("YARN资源管理器的逻辑服务名，由ZooKeeper管理HA");
        addProperty("yarn.resourcemanager.address", address);
        return this;
    }

    /**
     * 作业历史服务器高可用（可选，需配合JournalNode）
     * @param journalNodes JournalNode节点的地址，如：jn1:8485,jn2:8485
     */
    public MapReduceConfigBuilder journalNodes(String journalNodes) {
        addMessage("允许作业历史服务器在重启后恢复未完成的作业记录");
        addProperty("mapreduce.jobhistory.recovery.enable", "true");
        addMessage("使用JournalNode存储作业历史元数据，支持HA");
        addProperty("mapreduce.jobhistory.store.class", "org.apache.hadoop.mapreduce.jobhistory.JournalBasedStore");
        addMessage("JournalNode节点列表（与HDFS HA共享），用于存储作业历史元数据");
        addProperty("mapreduce.jobhistory.journal.nodes", journalNodes);
        return this;
    }

    /**
     * JobHistory Server RPC 地址（默认端口 10020）
     * 作业历史服务器配置（必选，用于查看作业历史）
     * @param address 如：node1:10020，作业历史服务器地址:端口
     */
    public MapReduceConfigBuilder jobHistoryAddress(String address) {
        addMessage("JobHistory Server RPC 地址");
        addProperty("mapreduce.jobhistory.address", address);
        return this;
    }

    /**
     * Web UI 访问地址（默认端口 19888）
     * 作业历史服务器的Web访问地址
     * @param webAddress 如：node1:19888
     */
    public MapReduceConfigBuilder jobHistoryWebAddress(String webAddress) {
        addMessage("Web UI 访问地址");
        addProperty("mapreduce.jobhistory.webapp.address", webAddress);
        return this;
    }

    /**
     * 已完成的作业历史存储目录（如 /data/mr-history/done）
     *
     * @param historyAddress 作业历史存储目录
     */
    public MapReduceConfigBuilder jobHistoryDoneDir(String historyAddress) {
        addMessage("已完成的作业历史存储目录");
        addProperty("mapreduce.jobhistory.done-dir", historyAddress);
        return this;
    }

    /**
     * 临时作业历史存储目录（如 /data/mr-history/tmp）
     *
     * @param historyTmpAddress 临时作业历史存储目录
     */
    public MapReduceConfigBuilder jobHistoryTmpDir(String historyTmpAddress) {
        addMessage("临时作业历史存储目录");
        addProperty("mapreduce.jobhistory.intermediate-done-dir", historyTmpAddress);
        return this;
    }

    /**
     * 启用 Map 输出压缩（减少网络传输）
     *
     * @param compress 是否压缩： true,false
     */
    public MapReduceConfigBuilder mapOutputCompress(String compress) {
        addMessage("启用 Map 输出压缩（减少网络传输）");
        addProperty("mapreduce.map.output.compress", compress);
        return this;
    }

    /**
     * 若启用 Kerberos，需添加以下配置（替换REALM为实际域名）
     *
     * @param realm 如：CMS.COM
     */
    public MapReduceConfigBuilder kerberosPrincipal(String host,String realm) {
        addMessage("jobTracker 节点添加 kerberos认证");
        addProperty("mapreduce.jobtracker.kerberos.principal", "mapred/"+host+"@"+realm);
        addMessage("taskTracker 节点添加 kerberos认证");
        addProperty("mapreduce.tasktracker.kerberos.principal", "mapred/"+host+"@"+realm);
        return this;
    }

    /**
     * 指定压缩算法）, 采用Snappy压缩
     */
    public MapReduceConfigBuilder mapOutputCompressCodec() {
        return mapOutputCompressCodec("org.apache.hadoop.io.compress.SnappyCodec");
    }

    /**
     * 指定压缩算法
     *
     * @param compressCodec 压缩格式
     */
    public MapReduceConfigBuilder mapOutputCompressCodec(String compressCodec) {
        addMessage("指定压缩算法");
        addProperty("mapreduce.map.output.compress.codec", compressCodec);
        return this;
    }

    /**
     * 单个 Mapper 任务内存上限（MB） 默认1024
     *
     * @param mapMemory 内存大小，单位MB
     */
    public MapReduceConfigBuilder mapMemory(String mapMemory) {
        addMessage("单个 Mapper 任务内存上限");
        addProperty("mapreduce.map.memory.mb", mapMemory);
        return this;
    }

    /**
     * 单个 Reducer 任务内存上限（MB）默认1024
     *
     * @param reduceMemory 内存大小，单位MB
     */
    public MapReduceConfigBuilder reduceMemory(String reduceMemory) {
        addMessage("单个 Reducer 任务内存上限");
        addProperty("mapreduce.reduce.memory.mb", reduceMemory);
        return this;
    }

    @Override
    public void addConfigHead() {
        configs.add("cat > $HADOOP_HOME/etc/hadoop/mapred-site.xml << EOF");
        super.addConfigHead();
    
    }

}
